
# - **统计维度**：按日/周/月
# - **指标需求**：
#   - 日均停留点数量分布
#   - 用户平均停留次数
#   - 停留点地理分布热力图
#   - 不同时段(早/中/晚)停留模式差异
# | 字段名      | 数据类型 | 描述     |
# | ----------- | -------- | -------- |
# | mdn         | STRING   | 手机号   |
# | date        | STRING   | 日期     |
# | county      | STRING   | 区县名称 |
# | lon         | DOUBLE   | 经度     |
# | lat         | DOUBLE   | 纬度     |
# | bsid        | STRING   | 基站ID   |
# | grid_id     | STRING   | 网格ID   |
# | biz_type    | STRING   | 业务类型 |
# | event_type  | STRING   | 事件类型 |
# | data_source | STRING   | 数据来源 |
from pyspark.sql import Window
from pyspark.sql.functions import *
from pyspark.sql.session import SparkSession
from pyspark.sql.session import SparkSession
from pyspark.sql.functions import *
spark= SparkSession.builder.appName('demo1_sparksession').enableHiveSupport().config("spark.jars", "jars/mysql-connector-java-8.0.29.jar").getOrCreate()

# /user/root/dm_db
staypoint_df = spark.read.format('csv').option('sep','\t').schema('mdn string,date string,county string,lon double,lat double,bsid string,grid_id string,biz_type string,event_type string,data_source string')\
    .load('/user/root/dm_db/staypoint')
#   - 日均停留点数量分布

staypoint_df = staypoint_df.withColumn('start_date',substring(split(col('date'),',')[1],1,8))
staypoint_df = staypoint_df.groupby(col('start_date'),col('county'))\
    .agg(count(col('county')))
staypoint_df.show(truncate=False)

staypoint_df.write \
    .format("jdbc") \
    .option("url", "jdbc:mysql://master:3306") \
    .option("driver", "com.mysql.cj.jdbc.Driver") \
    .option("dbtable", "dianxim.demo_1") \
    .option("user", "root") \
    .option("password", "123456") \
    .mode("overwrite") \
    .save()